package com.pw.study.flink.chapter9;

import com.pw.study.flink.entities.LoginEvent;
import com.pw.study.flink.entities.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;


import java.time.Duration;
import java.util.List;
import java.util.Map;

public class MyCepLogin {
    public static void main(String[] args) {
        login();
    }
    private static void login() {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        //输入数据
        SingleOutputStreamOperator<LoginEvent> data = env.readTextFile("data/file/LoginLog.csv").map(ads -> {
            String[] line = ads.split(",");
            return new LoginEvent(Long.valueOf(line[0]), line[1], line[2], Long.valueOf(line[3]) * 1000);
        });
        WatermarkStrategy<LoginEvent> strategy = WatermarkStrategy.<LoginEvent>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner((SerializableTimestampAssigner<LoginEvent>) (value, ts) -> value.getEventTime());

        SingleOutputStreamOperator<LoginEvent> stream = data.assignTimestampsAndWatermarks(strategy);

        Pattern<LoginEvent, LoginEvent> pattern = Pattern.<LoginEvent>begin("login").where(new SimpleCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent event) throws Exception {
                        return event.getEventType().equals("fail");
                    }
                }).times(2).consecutive()
                .within(Time.milliseconds(2001));
        PatternStream<LoginEvent> ps = CEP.pattern(stream, pattern);
        ps.select(new PatternSelectFunction<LoginEvent, String>() {
            @Override
            public String select(Map<String, List<LoginEvent>> pattern) throws Exception {
                Long userid = pattern.get("login").get(0).getUserId();
                String msg ="用户："+userid+" 正在恶意登录。。。。";
                return msg;
            }
        }).print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
